package com.ddxz.spark.read.tools;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import com.ddxz.spark.read.constant.CommonConstData;

/**
 * description
 *
 * @author 朱宗刚
 * @version 1.0.0
 * @date 2021/8/22
 */

public class HDFSTools {
    /**
     * HDFS接口初始化
     *
     * @return HDFS FileSystem
     * @throws IOException
     */
    private static FileSystem hdfsInit() throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", CommonConstData.defaultFs());
        FileSystem file = FileSystem.get(conf);
        return file;
    }

    public static boolean getfile(String hdfspath, String filepaht) throws IOException {
        try{
            FileSystem file = hdfsInit();
            file.copyToLocalFile(new Path(hdfspath), new Path(filepaht));
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    public static boolean mvfile(String oldhdfspath, String newhdfspath) throws IOException {
        try{
            FileSystem file = hdfsInit();
            file.rename(new Path(oldhdfspath), new Path(newhdfspath));
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    public static boolean deletefile(String hdfspath) throws IOException {
        try{
            FileSystem file = hdfsInit();
            file.delete(new Path(hdfspath),true);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }


    public static boolean fileExists(String filepath) throws IOException {
        FileSystem file = hdfsInit();
        if (file.exists( new Path(filepath))) {
            return true;
        } else {
            return false;
        }
    }

    public static void myMerge(String hdfs_path,String local_file) throws Exception{
        FileSystem fs =  hdfsInit();
        Path localPath = new Path(local_file);
        Path hdfsPath = new Path(hdfs_path);
        FileOutputStream fsdos = new FileOutputStream(localPath.toString());
        myWrite(hdfsPath,fs,fsdos);
        IOUtils.closeStream(fsdos);
    }

    public static void myWrite(Path path, FileSystem fs, OutputStream fsdos) throws Exception{
        FileStatus[] fst = fs.listStatus(path);
        FSDataInputStream fsdis = null;
        for(int i=0;i<fst.length;i++){
            if(fst[i].isDirectory()){
                myWrite(fst[i].getPath(),fs,fsdos);
            } else {
                fsdis = fs.open(fst[i].getPath());
                int read = 0;
                byte[] buffer = new byte[255];
                while((read=fsdis.read(buffer))>0){
                    fsdos.write(buffer, 0, read);
                }
                IOUtils.closeStream(fsdis);
            }
        }
    }
}
